package com.intct.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.func.DimBoradcatProcessFuncImpl;
import com.intct.utils.HBaseUtil;
import com.intct.utils.MysqlUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @author gufg
 * @since 2025-09-18 09:33
 */
public class DimMainAPI {
    public static void main(String[] args) throws Exception {
        //  1 配置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 从工程中resoucres路径读取配置文件
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(DimMainAPI.class.getResource("/intct.properties").getFile());

        //  2 配置数据源
        //  2.1 MySQL数据源
        String dbList = parameterTool.get("mysql.config.database", "travel_config");
        String tableList = parameterTool.get("mysql.dim.config.table", "travel_config.dim_table_config");
        MySqlSource<String> mysqlSource = MysqlUtil.getMysqlSource(parameterTool, dbList, tableList, StartupOptions.initial());
        DataStreamSource<String> mysqlSourceDS = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "dim_mysql_source");

//        mysqlSourceDS.print("mysql source >>> ");

        //  2.2 Kafka数据源
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(parameterTool.get("kafka.sink.bootstrap.servers", ""))
                .setTopics(Constant.KAFKA_DB_ODS_TOPIC_NAME)
                .setValueOnlyDeserializer(new SimpleStringSchema())
                // 主要是讲课使用earliest，生产环境使用commmit offset
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setGroupId("dim-main-app-group-1")
                .build();
        DataStreamSource<String> kafkaSourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dim_kafka_source");

//        kafkaSourceDS.print("kafka source >>> ");

        //  3 双流Join(connect算子 + Broadcast)

        //  Broadcast
        // 声明MapState状态
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor("dimMapState", String.class, String.class);

        // 将维度配置表广播
        BroadcastStream<String> broadcastStream = mysqlSourceDS.broadcast(mapStateDescriptor);

        // 主流(kafka)与广播流关联
        BroadcastConnectedStream<String, String> connectStream = kafkaSourceDS.connect(broadcastStream);

        // 关联流处理
        SingleOutputStreamOperator<String> processDS = connectStream.process(new DimBoradcatProcessFuncImpl(mapStateDescriptor));

        // 4 保存数据到HBase
        processDS.addSink(new RichSinkFunction<String>() {
            @Override
            public void invoke(String jsonStr, Context context) throws Exception {
                // {"rowkey":"511603","sink_family":"info","after":{"code":"511603","name":"四川省广安市前锋区"},"table":"dim_area_info"}
                JSONObject jsonObj = JSON.parseObject(jsonStr);

                // 表名
                String table = jsonObj.getString("table");
                String familyName = jsonObj.getString("sink_family");
                String namespace = "intct84";
                // 实现逻辑
                // 判断表是否，不存创建
                HBaseUtil.createTable(namespace, table, familyName);

                // 插入数据到HBase
                HBaseUtil.put(namespace, table, jsonObj);
            }
        });

        //  5 启动作业
        env.execute("dim_main_app");
    }
}
